package com.ssfeng.youxia.zk;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.nio.charset.Charset;

/**
 * Created by liyuhang on 2019-09-21.
 */
public class ZkClient {

    private static final Logger logger = LoggerFactory.getLogger(ZkClient.class);


    private String zookeeperConnectionString = "127.0.0.1:2181";

    private CuratorFramework client = null;

    private ZkClient() {
        RetryPolicy retryPolicy = new ExponentialBackoffRetry(1000, 3);
        client = CuratorFrameworkFactory.newClient(zookeeperConnectionString, retryPolicy);
        client.start();
    }


    public static final ZkClient getInstance() {
        return ZkClientHolder.INSTANCE;
    }

    public String getProperty(String type, String property) {
        String path = this.getPropertyPath(type, property);
        return getData(path);
    }

    public String getData(String path) {
        try {
            byte[] bytes = client.getData().forPath(path);
            if (bytes == null || bytes.length == 0) {
                return null;
            }
            String result = new String(bytes, Charset.forName("utf-8"));
            return result;
        } catch (Exception e) {
            e.printStackTrace();
        }
        return null;
    }

    public void addPropertyListener(String type,String property,ChangeListener changeListener) {
        String path = this.getPropertyPath(type, property);
        NodeCache nodeCache = new NodeCache(client, path);
        try {
            nodeCache.start(true);
        } catch (Exception e) {
            e.printStackTrace();
        }
        nodeCache.getListenable().addListener(new NodeCacheListener() {
            @Override
            public void nodeChanged() throws Exception {
                byte[] bytes = nodeCache.getCurrentData().getData();
                String result = new String(bytes, Charset.forName("utf-8"));
                changeListener.getData(result);
            }
        });

    }

    private String getPropertyPath(String type,String property){
        StringBuilder sb = new StringBuilder();
        sb.append("/configs/");
        sb.append(type);
        sb.append("/");
        sb.append(property);
        return sb.toString();
    }

    public void stop() {
        client.close();
    }

    private static class ZkClientHolder {
        private static final ZkClient INSTANCE = new ZkClient();
    }
}
